package com.algo.panelmn;
import java.io.IOException;
import java.util.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import java.util.regex.*;
import java.net.URLEncoder;
import java.net.URLDecoder;
import java.io.UnsupportedEncodingException;

public class format1 extends Configured implements Tool {
	public static class Mapper extends org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Text, Text> 
	{
		//LongWritable one = new LongWritable(1);
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String str=value.toString();
            if (str.indexOf("plt=0") >= 0)
            {
                //pca
                
			    int beg2=str.indexOf("uuid=");
			    if(beg2 == -1)
			    	return;
			    int end2=str.indexOf("^",beg2+5);
			    if(end2 == -1)
			    	return;
			    String uuid=str.substring(beg2+5,end2);
			    if(uuid.length() == 0)
			    	return;
			    
			    int beg=str.indexOf("c=");
			    String url= null;
			    if(beg==-1)
			    {
			    	int beg4 = str.indexOf("pr=");
			    	if (beg4 == -1)
			    		return;
			    	int end4 = str.indexOf("^",beg4 + 3);
			    	url = str.substring(beg4 + 3,end4);
			    	if(url.length() == 0)
			    		return;
			    }
			    else
			    {
			    	int end=str.indexOf("^",beg + 2);
			    	if(end==-1)
			    		return;
			    	url=str.substring(beg + 2,end);
			    	if(url.length() == 0)
			    		return;
			    }
			    String newurl= URLDecoder.decode(url,"UTF-8");
			    
			    String renren = "^http://(www\\.)?renren\\.com/([0-9]+)$";
			    String weibo = "^http://(www\\.)?weibo\\.com/([^/]+)/home";
			    String qq = "^http://user\\.qzone\\.qq\\.com/([0-9]+)/infocenter";
			    
				String re = uuid;
				if (Pattern.matches(renren,newurl))
				{
					//renren
					//context.write(new Text(uuid), new Text("r"));
					String p = newurl.replaceAll(renren, "$2");
					re += "^"+"r"+p; 
					//re += "^" 
				}
				else if (Pattern.matches(weibo,newurl))
				{
					//weibo
					String p = newurl.replaceAll(weibo, "$2");
					//context.write(new Text(uuid), new Text("w"));
					re += "^"+"w"+p;
				}
				else if (Pattern.matches(qq,newurl))
				{
					//qq
					String p = newurl.replaceAll(qq, "$1");
					//context.write(new Text(uuid), new Text("q"));
					re += "^"+"q"+p;
				}
				else
				{
					//other
					//context.write(new Text(uuid), new Text("o"));
					return;
				}
				
				int beg8 = str.indexOf("ti=");
			    if(beg8 == -1)
			    	return;
			    int end9=str.indexOf("^", beg8 + 3);
			    if(end9 == -1)
			    	return;
			    String ti = str.substring(beg8 + 3, end9);
			    if(ti.length() == 0)
		    		return;
				
			    re += "^"+ti;
			    
			    int beg10=str.indexOf("av=");
			    if(beg10 == -1)
			    	return;
			    
			    if (beg10+5 > str.length())
			    	return;
			    String av = str.substring(beg10 + 3, beg10 + 5);
			    
			    re += "^" + av;
			    
			    //fianl output
			    context.write(new Text(re), new Text(""));
            }
            
		}
	}
	public static class Reducer extends org.apache.hadoop.mapreduce.Reducer<Text, Text, Text, Text> 
	{
		public void reduce(Text key, Iterable<Text> values,Context context)
			 throws IOException, InterruptedException 
        {
			
            
			context.write(key, new Text(""));
				
		}
	}
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new format1(), args);
		System.exit(res);
	}

	public int run(String[] args) throws Exception {

		Configuration conf = getConf();

		@SuppressWarnings("deprecation")
		Job job = new Job(conf, "format1");
		job.setJarByClass(format1.class);
		job.setMapperClass(Mapper.class);
		job.setReducerClass(Reducer.class);
		job.setInputFormatClass(com.supertool.tong.merger.util.MzSequenceFileInputFormat.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		job.setNumReduceTasks(60);

		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		//FileInputFormat.addInputPath(job, new Path(args[0]));
		FileInputFormat.addInputPath(job, new Path(args[0]));
		return job.waitForCompletion(true) ? 0 : 1;
	}
}